package com.atguigu.app;

import com.atguigu.bean.IdCount;
import com.atguigu.bean.WaterSensor;
import org.apache.commons.compress.utils.Lists;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Iterator;

public class Flink09_02 {

    public static void main(String[] args) throws Exception {

        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        //2.读取端口数据
        SingleOutputStreamOperator<String> socketTextStream = env.socketTextStream("hadoop102", 9999)
                .assignTimestampsAndWatermarks(WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<String>() {
                    @Override
                    public long extractTimestamp(String element, long recordTimestamp) {
                        String[] split = element.split(",");
                        return Long.parseLong(split[1]) * 1000L;
                    }
                }));

        //3.将每行数据转换为JavaBean对象,提取时间戳生成WaterMark
        SingleOutputStreamOperator<WaterSensor> waterSensorDS = socketTextStream.map(line -> {
            String[] fields = line.split(",");
            return new WaterSensor(fields[0],
                    Long.parseLong(fields[1]),
                    Double.parseDouble(fields[2]));
        });

        waterSensorDS.print("waterSensorDS>>>>>");
//                .assignTimestampsAndWatermarks(WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
//            @Override
//            public long extractTimestamp(WaterSensor element, long recordTimestamp) {
//                return element.getTs() * 1000L;
//            }
//        }));

        //4.按照传感器ID分组
        KeyedStream<Tuple2<String, Integer>, String> keyedStream = waterSensorDS
                .map(new MapFunction<WaterSensor, Tuple2<String, Integer>>() {
                    @Override
                    public Tuple2<String, Integer> map(WaterSensor value) throws Exception {
                        return new Tuple2<>(value.getId(), 1);
                    }
                })
                .keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
                    @Override
                    public String getKey(Tuple2<String, Integer> value) throws Exception {
                        return value.f0;
                    }
                });

        //5.开窗
        WindowedStream<Tuple2<String, Integer>, String, TimeWindow> windowedStream = keyedStream
                .window(SlidingEventTimeWindows.of(Time.seconds(30), Time.seconds(5)))
                .allowedLateness(Time.seconds(5));

        //6.聚合,计算WordCount并添加窗口的信息
        SingleOutputStreamOperator<IdCount> aggregateDS = windowedStream.aggregate(new AggFunc(), new IdCountWindowFunc());
        aggregateDS.print("aggregateDS>>>>>>>>");

        //7.按照窗口信息重新分组
        KeyedStream<IdCount, Long> tsKeyedStream = aggregateDS.keyBy(IdCount::getTs);

        //8.ProcessFunction利用状态编程+定时器，完成最终的排序任务
        SingleOutputStreamOperator<String> result = tsKeyedStream.process(new TopNCountFunc(3));

        //9.打印结果
        result.print();

        //10.启动
        env.execute();

    }

    public static class AggFunc implements AggregateFunction<Tuple2<String, Integer>, Integer, Integer> {

        @Override
        public Integer createAccumulator() {
            return 0;
        }

        @Override
        public Integer add(Tuple2<String, Integer> value, Integer accumulator) {
            return accumulator + 1;
        }

        @Override
        public Integer getResult(Integer accumulator) {
            return accumulator;
        }

        @Override
        public Integer merge(Integer a, Integer b) {
            return a + b;
        }
    }

    public static class IdCountWindowFunc implements WindowFunction<Integer, IdCount, String, TimeWindow> {

        @Override
        public void apply(String key, TimeWindow window, Iterable<Integer> input, Collector<IdCount> out) throws Exception {

            //取出Count数据
            Integer ct = input.iterator().next();

            //获取窗口信息
            long windowEnd = window.getEnd();

            //写出数据
            out.collect(new IdCount(windowEnd, key, ct));

        }
    }

    public static class TopNCountFunc extends KeyedProcessFunction<Long, IdCount, String> {

        //定义属性
        private Integer topSize;

        public TopNCountFunc(Integer topSize) {
            this.topSize = topSize;
        }

        //定义状态
        private MapState<String, IdCount> mapState;

        @Override
        public void open(Configuration parameters) throws Exception {
            MapStateDescriptor<String, IdCount> mapStateDescriptor = new MapStateDescriptor<>("map-state", String.class, IdCount.class);

            mapStateDescriptor.enableTimeToLive(new StateTtlConfig.Builder(org.apache.flink.api.common.time.Time.seconds(10))
                    .setUpdateType(StateTtlConfig.UpdateType.Disabled)
                    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                    .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)
                    .build());

            mapState = getRuntimeContext().getMapState(mapStateDescriptor);
        }

        @Override
        public void processElement(IdCount value, Context ctx, Collector<String> out) throws Exception {

            //将数据加入状态
            mapState.put(value.getId(), value);

            //注册定时器,用于触发当数据收集完成时,排序输出
            TimerService timerService = ctx.timerService();
            timerService.registerEventTimeTimer(value.getTs() + 100L);

            //注册定时器用于触发状态清除
            timerService.registerEventTimeTimer(value.getTs() + 5000L + 100L);

        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {

            if (timestamp == ctx.getCurrentKey() + 5000L + 100L) {
                //清空状态
                mapState.clear();
                return;
            }

            //取出状态中的数据
            Iterator<IdCount> iterator = mapState.values().iterator();
            ArrayList<IdCount> idCounts = Lists.newArrayList(iterator);

            //排序
            idCounts.sort(new Comparator<IdCount>() {
                @Override
                public int compare(IdCount o1, IdCount o2) {
                    if (o1.getCt() > o2.getCt()) {
                        return -1;
                    } else if (o1.getCt() < o2.getCt()) {
                        return 1;
                    } else {
                        return 0;
                    }
                }
            });

            //输出
            StringBuilder sb = new StringBuilder("===================\n");
            for (int i = 0; i < Math.min(idCounts.size(), topSize); i++) {

                //取出数据
                IdCount idCount = idCounts.get(i);
                sb
                        .append("Top").append(i + 1)
                        .append(",ID:").append(idCount.getId())
                        .append(",CT:").append(idCount.getCt())
                        .append("\n");
            }
            sb.append("===================\n");
            out.collect(sb.toString());

            //清空状态
//            listState.clear();

        }
    }

}
